package org.pentaho.di.trans.steps.redis.output;

import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

import redis.clients.jedis.Jedis;
import redis.clients.util.Pool;

/**
 * The Redis Input step looks up value objects, from the given key names, from Redis server(s).
 */
public class RedisOutput extends BaseStep implements StepInterface {
	public static Class<?> PKG = RedisOutputMeta.class; // for i18n purposes, needed by Translator2!! $NON-NLS-1$

	protected RedisOutputMeta meta;
	protected RedisOutputData data;
	String valuefieldname ;
	String idfieldname ;
	String tablename;
	ArrayList<Object> rowkey=new ArrayList<Object>();
	public RedisOutput(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
			Trans trans) {
		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
	}

	private static Pool<Jedis> pool = null;
	//ExecutorService service = Executors.newFixedThreadPool(12);

	@Override
	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		long start = System.currentTimeMillis(); 
		if (super.init(smi, sdi)) {
			try {
				pool = ((RedisOutputMeta) smi).createJedisPool();
				long end = System.currentTimeMillis();  
				logBasic("建立连接池 毫秒："+(end-start));

				return true;
			} catch (Exception e) {
				logError(BaseMessages.getString(PKG, "RedisInput.Error.ConnectError"), e);
				return false;
			}
		} else {
			return false;
		}


	}

	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
		meta = (RedisOutputMeta) smi;
		data = (RedisOutputData) sdi;

		// TODO Auto-generated method stub
		Jedis jedis = pool.getResource();
		Object[] r = getRow(); // get row, set busy!
	
		// If no more input to be expected, stop
		if (r == null) {
			setOutputDone();
			return false;
		}

		if (first) {
			first = false;
			// clone input row meta for now, we will change it (add or set inline) later
			data.outputRowMeta = getInputRowMeta().clone();
			// Get output field types
			meta.getFields(data.outputRowMeta, getStepname(), null, null, this);
		}

		int idFieldIndex = getInputRowMeta().indexOfValue(meta.getIdFieldName());
		int valueFieldIndex = getInputRowMeta().indexOfValue(meta.getValueFieldName());
		if ((idFieldIndex < 0) || (valueFieldIndex < 0)) {
			throw new KettleException(BaseMessages.getString(PKG, "RedisInputMeta.Exception.FieldNameNotFound"));
		}

		if (null != r[idFieldIndex]) {
			String key = (String)r[idFieldIndex];
			if (key.length() > 0) {
				jedis.set(key, (String)r[valueFieldIndex]);
			}
		}

		putRow(data.outputRowMeta, r); // copy row to possible alternate rowset(s).

//		RedisOutputThread thread=new RedisOutputThread(this, jedis, r);
//		service.submit(thread);

		if (checkFeedback(getLinesRead())) {
			if (log.isBasic()) {
				logBasic(BaseMessages.getString(PKG, "RedisOutput.Log.LineNumber") + getLinesRead());
			}
		}
		return true;
	}

	@Override
	public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
		super.dispose(smi, sdi);
		if(pool!=null){
			pool.close();
			pool.destroy();
		}
		//service.shutdown();
	}
}
